package com.xingzhengfan.eurekaclient;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration
public class RabbitMQConfig {

    private static final Logger logger = LoggerFactory.getLogger(RabbitMQConfig.class);

    @Autowired
    private MQConsumerListener mqConsumerListener;

    @Bean(name = "amqpTemplate")
    public RabbitTemplate amqpTemplate() {
        RabbitTemplate template = new RabbitTemplate(rabbitmqConnectionFactory());
        template.setMessageConverter(objectMessageConverter());
        return template;
    }
    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(100);
        executor.setMaxPoolSize(200);
        executor.setQueueCapacity(2000);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        return executor;
    }
    @Bean(name = "rabbitmqConnectionFactory")
    public CachingConnectionFactory rabbitmqConnectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory("210.22.173.210", 25672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        return factory;
    }
    @Bean(name = "objectMessageConverter")
    public ObjectMessageConverter objectMessageConverter() {
        ObjectMessageConverter converter = new ObjectMessageConverter();
        return converter;
    }
    @Bean
    public Queue toFinanceQueue() {
        return new Queue(QoConstants.FINANCE_QUEUE_NAME);
    }
    @Bean
    public Queue toOrderQueue() {
        return new Queue(QoConstants.ORDER_QUEUE_NAME);
    }
    @Bean
    public FanoutExchange toFinanceExchange() {
        return new FanoutExchange(QoConstants.FINANCE_EXCHANGE_NAME, true, false);
    }
    @Bean
    public FanoutExchange toOrderExchange() {
        return new FanoutExchange(QoConstants.ORDER_EXCHANGE_NAME, true, false);
    }
    @Bean
    public Binding toFinanceBinding() {
        return BindingBuilder.bind(toFinanceQueue()).to(toFinanceExchange());
    }
    @Bean
    public Binding toOrderBinding() {
        return BindingBuilder.bind(toOrderQueue()).to(toOrderExchange());
    }
    @Bean(name = "simpleMessageListenerContainer")
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(rabbitmqConnectionFactory());
        simpleMessageListenerContainer.setMessageConverter(objectMessageConverter());
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
        simpleMessageListenerContainer.setConcurrency("5");
        Queue[] queues = {toFinanceQueue(), toOrderQueue()};
        simpleMessageListenerContainer.setQueues(queues);
        simpleMessageListenerContainer.setMessageListener(msg -> {
            logger.info("messageBody:[{}]", JSON.toJSONString(msg));
            if (QoConstants.FINANCE_QUEUE_NAME.equals(msg.getMessageProperties().getConsumerQueue())) {
                logger.info("FINANCE receive a message:[{}]", JSON.toJSONString(msg));
                mqConsumerListener.financeConsumer(msg);
            }
            if (QoConstants.ORDER_QUEUE_NAME.equals(msg.getMessageProperties().getConsumerQueue())) {
                logger.info("ORDER receive a message:[{}]", JSON.toJSONString(msg));
                mqConsumerListener.orderConsumer(msg);
            }
        });
        return simpleMessageListenerContainer;
    }
}
